package com.gupao.javavip.rpc.framework.client;

import lombok.extern.apachecommons.CommonsLog;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

@CommonsLog
public class ZKDiscovery implements IDiscovery {
    @Autowired
    private RPCClientConfig config;

    @Autowired
    private ForeachLoadBalanceStrategy foreachLoadBalanceStrategy;

    public void setConfig(RPCClientConfig config) {
        this.config = config;
    }

    public void setForeachLoadBalanceStrategy(ForeachLoadBalanceStrategy foreachLoadBalanceStrategy) {
        this.foreachLoadBalanceStrategy = foreachLoadBalanceStrategy;
    }

    private CuratorFramework curatorFramework = null;

    private volatile List<String> serviceRepos = new ArrayList<>();

    @Override
    public ServiceNode discovery(String serviceId) {
        if (StringUtils.isEmpty(serviceId)) {
            ServiceNode node = new ServiceNode();
            node.setHost(config.getHost());
            node.setPort(config.getPort());
            return node;
        }

        if (curatorFramework == null) {
            initCuratorFramework();
        }

        return doDiscovery(serviceId);
    }

    private void initCuratorFramework() {
        curatorFramework = CuratorFrameworkFactory.builder().
                connectString(config.getRegistryAdd()).sessionTimeoutMs(5000).
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).
                namespace("registry")
                .build();
        curatorFramework.start();
    }

    private ServiceNode doDiscovery(String serviceId) {
        String path = "/" + serviceId;
        if (serviceRepos.isEmpty()) {
            try {
                serviceRepos = curatorFramework.getChildren().forPath(path);
                registryWatch(path);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        return foreachLoadBalanceStrategy.selectHost(serviceRepos);
    }

    private void registryWatch(final String path) throws Exception {
        PathChildrenCache nodeCache = new PathChildrenCache(curatorFramework, path, true);
        PathChildrenCacheListener nodeCacheListener = (curatorFramework1, pathChildrenCacheEvent) -> {
            log.info("服务节点变更，变更前：" + serviceNodeStr());
            serviceRepos = curatorFramework1.getChildren().forPath(path);// 再次更新本地的缓存地址
            log.info("最新的服务节点：" + serviceNodeStr());
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start();

    }

    private String serviceNodeStr() {
        StringBuilder stringBuilder = new StringBuilder();
        for (String node : serviceRepos) {
            stringBuilder.append(node).append(",");
        }
        return stringBuilder.toString();
    }
}
